#include "Server.h"

static swConnection* swServer_connection_new(swServer *serv, swListenPort *ls, int fd, int from_fd, int reactor_id);

swServerG SwooleG;
swWorkerG SwooleWG;

static int swServer_start_proxy(swServer *serv);

void swServer_init(swServer *serv){
	bzero(serv,sizeof(swServer));

	serv->gs = SwooleG.memory_pool->alloc(SwooleG.memory_pool, sizeof(swServerGS));
    if (serv->gs == NULL)
    {
        printf("[Master] Fatal Error: failed to allocate memory for swServer->gs.");
    }

	serv->max_connection = 1024;
	serv->worker_num = SwooleG.cpu_num;
	serv->reactor_num = SwooleG.cpu_num;

    SwooleG.serv = serv;
}

int swServer_master_onAccept(swReactor *reactor, swEvent *event){
	printf("accept run!\n");

	struct sockaddr_in client_addr;
	swServer *serv = reactor->ptr;
	swReactor *sub_reactor;
	socklen_t client_addrlen = sizeof(client_addr);
	int new_fd = 0, i,reactor_id = 0;
	swListenPort *listen_host = serv->connection_list[event->fd].object;

	for (i = 0; i < SW_ACCEPT_MAX_COUNT; i++)
    {
#ifdef HAVE_ACCEPT4
		printf("accept4\n");
        new_fd = accept4(event->fd, (struct sockaddr *) &client_addr, &client_addrlen, SOCK_NONBLOCK | SOCK_CLOEXEC);
#else
		printf("accept\n");
        new_fd = accept(event->fd, (struct sockaddr *) &client_addr, &client_addrlen);
#endif
        if (new_fd < 0)
        {
            switch (errno)
            {
            case EAGAIN:
                return 0;
            case EINTR:
                continue;
            default:
                return 0;
            }
        }
#ifndef HAVE_ACCEPT4
        else
        {
            swoole_fcntl_set_option(new_fd, 1, 1);
        }
#endif

		reactor_id = new_fd % serv->reactor_num;
		//conn not write
		swConnection *conn = swServer_connection_new(serv, listen_host, new_fd, event->fd, reactor_id);
		//
		sub_reactor = &serv->reactor_threads[reactor_id].reactor;

		conn->connect_notify = 1;
		if(sub_reactor->add(sub_reactor,new_fd,SW_FD_TCP | SW_EVENT_WRITE) < 0){
			close(new_fd);
			return 0;
		}
	}


	return 0;
}

swListenPort* swServer_add_port(swServer *serv, char *host, int port){
	swListenPort *ls = SwooleG.memory_pool->alloc(SwooleG.memory_pool,sizeof(swListenPort));
	if(ls == NULL){
		return NULL;
	}

	swPort_init(ls);
	ls->port = port;
	strncpy(ls->host,host,strlen(host) + 1);

	int sock = swSocket_create();
	if(sock < 0){
		return NULL;
	}

	if(swSocket_bind(sock,ls->host,&ls->port) < 0){
		close(sock);
		return NULL;
	}

	swoole_fcntl_set_option(sock, 1, 1);

	ls->sock = sock;

	serv->listen_list = ls;

	return ls;
}

int swServer_create(swServer *serv){
	SwooleG.factory = &serv->factory;
	serv->factory.ptr = serv;

	//create session
	serv->session_list = sw_shm_calloc(SW_SESSION_LIST_SIZE, sizeof(swSession));

	return swReactorThread_create(serv);
}

static int swServer_start_proxy(swServer *serv){
	int ret;
    swReactor *main_reactor = SwooleG.memory_pool->alloc(SwooleG.memory_pool, sizeof(swReactor));

    ret = swReactor_create(main_reactor, SW_REACTOR_MAXEVENTS);
    if (ret < 0)
    {
        return -1;
    }

	main_reactor->thread = 1;
	main_reactor->socket_list = serv->connection_list;

	//add signal fd to events
	swSignalfd_setup(main_reactor);

	if(swPort_listen(serv->listen_list) < 0){
		return -1;
	}

	/**
	 * create reactor thread
	 */
	ret = swReactorThread_start(serv, main_reactor);
	if(ret < 0){
		return -1;
	}

	/**
	 * master thread loop
	 */

	SwooleG.main_reactor = main_reactor;
	SwooleG.pid = getpid();

	main_reactor->ptr = serv;
	main_reactor->setHandle(main_reactor, SW_FD_LISTEN, swServer_master_onAccept);

	//serv->onStart()
	
	return main_reactor->wait(main_reactor,NULL);
}

void swServer_store_listen_socket(swServer *serv){
	swListenPort *ls;
	int sockfd;

	ls = serv->listen_list;
	sockfd = ls->sock;
	serv->connection_list[sockfd].fd = sockfd;
	serv->connection_list[sockfd].object = ls;
}

int swServer_start(swServer *serv){
	swFactory *factory = &serv->factory;
	int ret;

	serv->workers = SwooleG.memory_pool->alloc(SwooleG.memory_pool, serv->worker_num * sizeof(swWorker));
	if(serv->workers == NULL){
		return -1;
	}

	serv->gs->event_workers.workers = serv->workers;
    serv->gs->event_workers.worker_num = serv->worker_num;

    int i;
    for (i = 0; i < serv->worker_num; i++)
    {
        serv->gs->event_workers.workers[i].pool = &serv->gs->event_workers;
    }

	if(factory->start(factory) < 0){
		return -1;
	}

	//signal init
	swServer_signal_init(serv);

	ret = swServer_start_proxy(serv);

	swServer_free(serv);

	return 0;
}

int swServer_free(swServer *serv){
	//pthread_join

	serv->factory.shutdown(&serv->factory);
	swReactorThread_free(serv);

	close(serv->listen_list->sock);
}

int swServer_tcp_notify(swServer *serv, swConnection *conn, int event){
	swDataHead notify_event;

	//fill notify_event
	notify_event.type = event;
	notify_event.from_id = conn->from_id;
	notify_event.fd = conn->fd;
	notify_event.from_fd = conn->from_fd;
	notify_event.len = 0;

	return serv->factory.notify(&serv->factory,&notify_event);
}

void swServer_store_pipe_fd(swServer *serv, swPipe *p){
	int master_fd = p->getFd(p,SW_PIPE_MASTER);

	serv->connection_list[p->getFd(p, SW_PIPE_WORKER)].object = p;
	serv->connection_list[master_fd].object = p;
}

static swConnection* swServer_connection_new(swServer *serv, swListenPort *ls, int fd, int from_fd, int reactor_id){
	swConnection* connection = NULL;
	int i;
	swSession *session;

	connection = &(serv->connection_list[fd]);
	bzero(connection, sizeof(swConnection));
	connection->fd = fd;
	connection->from_id = reactor_id;
	connection->from_fd = from_fd;
	connection->active = 1;

	uint32_t session_id = serv->gs->session_round;

	//get session id
    for (i = 0; i < serv->max_connection; i++)
    {
        session_id++;
        //SwooleGS->session_round just has 24 bits size;
        if (session_id == 1 << 24)
        {
            session_id = 1;
        }
        session = swServer_get_session(serv, session_id);
        //vacancy
        if (session->fd == 0)
        {
            session->fd = fd;
            session->id = session_id;
            session->reactor_id = connection->from_id;
            break;
        }
    }
    serv->gs->session_round = session_id;

	connection->session_id = session_id;

	return connection;
}

int swServer_tcp_send(swServer *serv, int fd, void *data, uint32_t length){
	swSendData _send;
	swFactory *factory = &(serv->factory);

	//do not think of big package

	_send.info.fd = fd;
	_send.info.type = SW_EVENT_TCP;
	_send.data = data;
	_send.info.len = length;
	_send.length = 0;
	return factory->finish(factory, &_send);
}

static void swServer_signal_handler(int sig){
	swServer *serv = SwooleG.serv;
	pid_t pid;
	int status;

	switch(sig){
		case SIGTERM:
			if(SwooleG.main_reactor){
				SwooleG.main_reactor->running = 0;
			}else{
				SwooleG.running = 0;
			}
			printf("Server is shutdown now.\n");
			break;
		case SIGCHLD:
			if(!SwooleG.running){
				break;
			}
			pid = waitpid(-1,&status,WNOHANG);
			if(pid > 0 && pid == serv->gs->manager_pid){
				printf("Fatal Error: manager process exit. status=%d,signal=%d.\n",WEXITSTATUS(status), WTERMSIG(status));
			}
			break;
		case SIGUSR1:
		case SIGUSR2:
			printf("kill manager process\n");
			kill(serv->gs->manager_pid,sig);
			break;
		default:
			break;
	}
}

void swServer_signal_init(swServer *serv){
	swSignal_add(SIGUSR1, swServer_signal_handler);
	swSignal_add(SIGUSR2, swServer_signal_handler);
	swSignal_add(SIGTERM, swServer_signal_handler);
	swSignal_add(SIGCHLD, swServer_signal_handler);
}
